package com.sunday.common.mq.rocket.brave.study.base.e12_POP;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;

import java.util.Collection;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 *
 */
@Slf4j
public class PullConsumerThread extends Thread {

    private DefaultLitePullConsumer consumer;

    private String node;

    private AtomicInteger count = new AtomicInteger();

    public PullConsumerThread(String namesrvAddr, String consumerGroup, String topic, String node) throws MQClientException {
        this.node = node;
        this.consumer = new DefaultLitePullConsumer(consumerGroup);
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.subscribe(topic, "*");
        //每次提取的最大消息数。
        consumer.setPullBatchSize(1);
        //自动消息确认
//        consumer.setAutoCommit(false);
        //实例名称
        // springboot服务配置了多个consumer，但没有指定instanceName。确保instanceName唯一就可以，consumer的实例名，在同一个group中保证唯一，可以用Topic名。
//        consumer.setInstanceName("user_pull_consumer");
        consumer.start();
        log.info("Consumer Started. {}", node);
//        Collection<MessageQueue> queues = consumer.fetchMessageQueues(topic);
//        log.info("topic queues size {}", queues.size());
//        queues.stream().forEach(queue -> {
//            log.info("Consume from the queue : {}", queue);
//        });
    }

    @Override
    public void run() {
        try {
            while (true) {
                List<MessageExt> messages = consumer.poll();
//                log.info("data size : {}", messages.size());
                messages.stream().forEach(messageExt -> log.info("{} -> {} -> {}", this.node, new String(messageExt.getBody()), messageExt));
                count.incrementAndGet();
            }
        } catch (Exception e) {
            log.error(e.toString(), e);
        }

    }

}
